package instance;


import data.handle.*;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

//实例的数据处理中枢
public class LocalInstance {

    //保存开启的几个数据处理线程，每个线程会处理一个分段的零件号
    private static HandlerThread[] threads;
    //负责处理part表的数据，每个处理线程会有一个PartHandle对象，建立数据结构保存part表的关键数据
    private static PartHandle[] partHandles;
    //负责处理lineItem的数据，每个处理线程会有一个LineItemHandle对象，建立数据结构保存lineItem表的关键数据
    private static LineItemHandle[] lineItemHandles;
    //线程池，用于查询结果的获取
    private static ExecutorService executorService;
    //记录实例加载数据的时间
    public volatile static long runningTime;

    /*
     * @Description: 开始启动数据处理线程
     * @param num : 开启的数据处理线程的数量
     * @param fileReaderThreadNum : 文件读取线程的数量
     * @return: {@link void}
     * @author: xff
     * @Date: 2022/7/5
     */
    public static void startLocalThreads(int num, int fileReaderThreadNum){

        threads = new HandlerThread[num];
        partHandles = new PartHandle[num];
        lineItemHandles = new LineItemHandle[num];

        int partKeyNum = 2000000 / (num * 4);//每个数据处理线程需要处理的零件号的数量
        for (int i = 0; i < num; i++) {
            partHandles[i] = new PartHandle(partKeyNum);
            //由于两个表之间存在连接查询，需要将part表的对象与lineItem表的对象关联起来
            lineItemHandles[i] = new LineItemHandle(partKeyNum, partHandles[i]);
        }

        long beginTimeMillis = System.currentTimeMillis();//记录实例的开始加载时间
        AtomicLong counterFile1 = new AtomicLong(0);//记录实例part表的数据加载行数
        AtomicLong counterFile2 = new AtomicLong(0);//记录实例lineItem表的数据加载行数
        CyclicBarrier cyclicBarrier = new CyclicBarrier(num, new Runnable() {

            @Override
            public void run() {
                runningTime = System.currentTimeMillis() - beginTimeMillis;
                System.out.println("use time: " + runningTime + "ms");
                System.out.println("Part all line: " + counterFile1.get());
                System.out.println("LineItem all line: " + counterFile2.get());
            }
        });

        //启动数据处理线程
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new HandlerThread(fileReaderThreadNum, partHandles[i], lineItemHandles[i]);
            threads[i].setTest(counterFile1, counterFile2, cyclicBarrier);
            new Thread(threads[i]).start();
        }
    }

    //获得当前实例上的销售总和
    public static long getQuantitySum(){
        long sum = 0;
        for (LineItemHandle lineItemHandle : lineItemHandles) {
            sum += lineItemHandle.getQuantitySum();
        }
        return sum;
    }

    /*
     * @Description: 查询指定品牌的零件销售量小于quantity数量的零件号和销售额
     * @param brandIndex : 品牌号
     * @param quantity : 销量
     * @return: {@link java.util.concurrent.LinkedBlockingQueue<data.handle.Result>}
     * @author: xff
     * @Date: 2022/7/5
     */
    public static LinkedBlockingQueue<Result> getResult(byte brandIndex, double quantity){
        LinkedBlockingQueue<Result> results = new LinkedBlockingQueue<>();
        if(executorService == null)
            executorService = Executors.newFixedThreadPool(threads.length);
        CountDownLatch downLatch = new CountDownLatch(threads.length);
        //多线程查询结果
        for (LineItemHandle lineItemHandle : lineItemHandles) {
            executorService.execute(new SelectResult(lineItemHandle, brandIndex, quantity, results, downLatch));
        }
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return results;
    }

    /*
     * @Description: 负责将数据包分配给对应范围的线程进行处理
     * @param packMessage : 封装了相关数据的数据包
     * @param index : 负责处理数据包的线程序号
     * @param fileIndex  : 文件序号
     * @return: {@link void}
     * @author: xff
     * @Date: 2022/7/5
     */
    public static void addData(PackMessage packMessage, int index, int fileIndex){
        threads[index].addData(packMessage, fileIndex);
    }
}
